package com.gome.ocean.service.thread;

import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.quartz.Scheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import com.gome.ocean.common.enums.State;
import com.gome.ocean.dao.model.datax.JobBO;
import com.gome.ocean.service.datax.JobService;
import com.gome.ocean.service.task.OneWayTask;

/**
 * 线程管理类
 * 
 * @author haqiaolong 2016年7月27日 下午2:22:17
 * @since 5.0.0
 */
@Service
public class ThreadFactoryImpl implements ThreadFactory, InitializingBean, DisposableBean {

    private static final Logger                       LOG      = LoggerFactory.getLogger(ThreadFactoryImpl.class);

    private static ConcurrentHashMap<Long, Future<?>> allTask  = new ConcurrentHashMap<Long, Future<?>>();

    private final ScheduledExecutorService            schedule = Executors.newScheduledThreadPool(1);
    
    @Autowired
    private JobService          jobService;

    @Autowired
    private ThreadPoolTaskExecutor                    taskExecutor;

    @Autowired
    @Qualifier("scheduler")
    private Scheduler                                 scheduler;

    @Override
    public void stop(Long jobId) {
        if (allTask.containsKey(jobId)) {
            Future<?> future = allTask.get(jobId);
            if (!future.isDone() && !future.isCancelled()) {
                future.cancel(true);
                allTask.remove(jobId);
                LOG.info("任务终止完成！jobId=" + jobId);
                return;
            }
        }
        LOG.warn("没有找到该任务！jobId=" + jobId);
    }

    @Override
    public void start(OneWayTask jobTask) {
        Future<?> future = taskExecutor.submit(jobTask);
        if (!allTask.containsKey(jobTask.getId())) {
            allTask.put(jobTask.getId(), future);
            LOG.info("任务启动成功！jobId=" + jobTask.getId());
        }
    }

    @Override
    public void destroy() throws Exception {
        taskExecutor.shutdown();
        schedule.shutdown();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 定时清除无用future
        schedule.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                for (Entry<Long, Future<?>> task : allTask.entrySet()) {
                    if (task.getValue().isDone()) {
                    	JobBO job = jobService.findJob(task.getKey());
                    	if(job != null){
                    		if(State.SUCCESS == State.getState(job.getState())){
                    			allTask.remove(task.getKey());
                                LOG.info("任务已完成！移除任务！jobId=" + task.getKey());
                    		}else{
                    			LOG.info("任务正在运行！忽略！jobId=" + task.getKey());
                    		}
                    	}
                        
                    }
                }
            }
        }, 0, 60, TimeUnit.SECONDS);
    }

	@Override
	public boolean exist(Long jobId) {
		return allTask.containsKey(jobId);
	}
}
